package com.dataconver.sync;

import com.dataconver.config.ConfigManager;
import com.dataconver.db.DatabaseManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Calendar;
import java.util.TimeZone;

/**
 * wm_mulity_route_meter_data 关联增量同步
 * - 根据 wm_mulity_route_his 的新增记录来同步对应的仪表数据
 * - 使用 wm_sync_status 表中 wm_mulity_route_his 的最后更新时间作为条件
 * - 查询新增记录ID，然后根据这些ID查询对应的仪表数据
 */
public class RelatedMeterSyncStrategy implements SyncStrategy {
    private static final Logger logger = LoggerFactory.getLogger(RelatedMeterSyncStrategy.class);

    private final String sourceTable = "MulityRoutMeterData";
    private final String targetTable = "wm_mulity_route_meter_data";
    private final int batchSize;

    public RelatedMeterSyncStrategy() {
        this.batchSize = ConfigManager.getInt("sync.batchSize", 1000);
    }

    @Override
    public SyncResult execute(String tableName) {
        SyncResult result = new SyncResult();
        result.setTableName(targetTable);
        result.setStartTime(System.currentTimeMillis());
        

        
        try {
            // 1) 获取 wm_mulity_route_his 表中新增记录的ID列表
            List<Integer> newHisIds = getNewHisIds();
            if (newHisIds.isEmpty()) {
    
                result.setSuccess(true);
                result.setEndTime(System.currentTimeMillis());
                return result;
            }
            


            // 2) 根据新增的ID查询对应的仪表数据
            int processed = syncMeterData(newHisIds);

            // 3) 同步状态更新将在所有表同步完成后统一处理


            result.setProcessedRows(processed);
            result.setTotalRows(processed);
            result.setInsertedRows(processed);
            result.setUpdatedRows(0);
            result.setSuccess(true);
            result.setEndTime(System.currentTimeMillis());
            

            
        } catch (Exception e) {
            logger.error("wm_mulity_route_meter_data 关联增量同步失败", e);
            result.setSuccess(false);
            result.setErrorMessage(e.getMessage());
            result.setEndTime(System.currentTimeMillis());
        }
        return result;
    }

    /**
     * 获取 wm_mulity_route_his 表中新增记录的ID列表
     */
    private List<Integer> getNewHisIds() throws SQLException {
        List<Integer> ids = new ArrayList<>();
        
        // 获取 wm_mulity_route_his 表的最后同步时间
        Timestamp lastSyncTime = getLastSyncTime("wm_mulity_route_his");
        if (lastSyncTime == null) {

            return ids;
        }
        
        // 查询新增的历史记录ID
        String sql = "SELECT DISTINCT h.`id` FROM `wm_mulity_route_his` h " +
                    "WHERE h.`imei` IN (SELECT DISTINCT `imei` FROM `wm_user_rtu`) " +
                    "AND h.`systime` > ?";
        
        try (Connection conn = DatabaseManager.getMysqlConnection();
             PreparedStatement ps = conn.prepareStatement(sql)) {
            ps.setTimestamp(1, lastSyncTime);
            
            
            
            try (ResultSet rs = ps.executeQuery()) {
                while (rs.next()) {
                    ids.add(rs.getInt(1));
                }
            }
        }
        

        return ids;
    }

    /**
     * 同步仪表数据
     */
    private int syncMeterData(List<Integer> hisIds) throws SQLException {
        int processed = 0;
        int inserted = 0;
        int updated = 0;

        try (Connection srcConn = DatabaseManager.getSqlserverConnection();
             Connection dstConn = DatabaseManager.getMysqlConnection()) {

            dstConn.setAutoCommit(false);

            // 按批次处理，避免IN过长
            for (int start = 0; start < hisIds.size(); start += 1000) {
                int end = Math.min(start + 1000, hisIds.size());
                List<Integer> batchHis = hisIds.subList(start, end);

                // 查询SQL：根据 hisid 查询仪表数据
                StringBuilder select = new StringBuilder();
                select.append("SELECT [id],[hisid],[metercode],[routeid],[sequence],[data],[status],[mulit],[typecode],[strong] FROM [").append(sourceTable).append("] WHERE ");
                select.append("[hisid] IN (");
                for (int i = 0; i < batchHis.size(); i++) {
                    if (i > 0) select.append(",");
                    select.append("?");
                }
                select.append(") ORDER BY [id] ASC");

                // UPSERT SQL
                String upsert = "INSERT INTO `" + targetTable + "` (`id`,`hisid`,`metercode`,`routeid`,`sequence`,`data`,`status`,`mulit`,`typecode`,`strong`) " +
                        "VALUES(?,?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE " +
                        "`hisid`=VALUES(`hisid`),`metercode`=VALUES(`metercode`),`routeid`=VALUES(`routeid`)," +
                        "`sequence`=VALUES(`sequence`),`data`=VALUES(`data`),`status`=VALUES(`status`)," +
                        "`mulit`=VALUES(`mulit`),`typecode`=VALUES(`typecode`),`strong`=VALUES(`strong`)";

                

                try (PreparedStatement ps = srcConn.prepareStatement(select.toString(), ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
                     PreparedStatement up = dstConn.prepareStatement(upsert)) {
                    
                    // 设置查询参数
                    int idx = 1;
                    for (Integer hisId : batchHis) {
                        ps.setInt(idx++, hisId);
                    }

                    try (ResultSet rs = ps.executeQuery()) {
                        int batch = 0;
                        while (rs.next()) {
                            // 设置UPSERT参数
                            up.setLong(1, rs.getLong("id"));
                            up.setInt(2, rs.getInt("hisid"));
                            up.setString(3, rs.getString("metercode"));
                            up.setInt(4, rs.getInt("routeid"));
                            up.setInt(5, rs.getInt("sequence"));
                            up.setBigDecimal(6, rs.getBigDecimal("data"));
                            up.setInt(7, rs.getInt("status"));
                            up.setInt(8, rs.getInt("mulit"));
                            up.setInt(9, rs.getInt("typecode"));
                            up.setInt(10, rs.getInt("strong"));

                            up.addBatch();
                            batch++;
                            processed++;

                            if (processed % 1000 == 0) {
                                logger.info("wm_mulity_route_meter_data 已处理 {} 条", processed);
                            }

                            if (batch >= batchSize) {
                                int[] res = up.executeBatch();
                                dstConn.commit();
                                for (int r : res) {
                                    if (r > 1) updated++; else inserted++;
                                }
                                batch = 0;
                            }
                        }

                        if (batch > 0) {
                            int[] res = up.executeBatch();
                            dstConn.commit();
                            for (int r : res) {
                                if (r > 1) updated++; else inserted++;
                            }
                        }
                    }
                }
            }

            dstConn.setAutoCommit(true);
        }
        

        return processed;
    }

    /**
     * 获取指定表的最后同步时间
     */
    private Timestamp getLastSyncTime(String tableName) throws SQLException {
        String sql = "SELECT last_sync_time FROM wm_sync_status WHERE table_name = ?";
        try (Connection conn = DatabaseManager.getMysqlConnection();
             PreparedStatement ps = conn.prepareStatement(sql)) {
            ps.setString(1, tableName);
            try (ResultSet rs = ps.executeQuery()) {
                if (rs.next()) {
                    Timestamp ts = rs.getTimestamp(1);
                    if (ts != null) {
                        // 转换为北京时间显示
                        Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("Asia/Shanghai"));
                        cal.setTime(ts);
                        Timestamp beijingTime = new Timestamp(cal.getTimeInMillis());
                        

                        return ts; // 返回原始值用于数据库查询
                    }
                }
            }
        }

        return null;
    }



    @Override
    public String getStrategyName() {
        return "RELATED_INCREMENTAL";
    }
}
